package cn.com.dhcc.turbo.service.registry.sync;

import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import cn.com.dhcc.app.core.exception.ServiceException;
import cn.com.dhcc.app.core.handler.threadpoolhandler.ThreadPool;
import cn.com.dhcc.app.core.service.Log;
import cn.com.dhcc.app.core.service.Log.LOGTYPE;
import cn.com.dhcc.app.core.util.JsonUtil;
import cn.com.dhcc.app.core.util.UUIDGenerator;
import cn.com.dhcc.app.pub.core.consts.AppIocBean;
import cn.com.dhcc.app.pub.core.consts.NodeInfoParam;
import cn.com.dhcc.turbo.entity.registry.NodeInfo;
import cn.com.dhcc.turbo.entity.registry.NodePipeline;
import cn.com.dhcc.turbo.entity.registry.SynchronizeInfo;
import cn.com.dhcc.turbo.entity.registry.NodePipeline.PIPE_TYPE;
import cn.com.dhcc.turbo.entity.registry.SynchronizeInfo.SyncOperType;
import cn.com.dhcc.turbo.entity.registry.SynchronizeInfo.SyncStatus;
import cn.com.dhcc.turbo.entity.registry.SynchronizeInfo.SyncType;
import cn.com.dhcc.turbo.service.queue.tlq.TlqCache;
import cn.com.dhcc.turbo.service.registry.NodePipelineService;

/**
 * 同步中心
 * @author qiao
 * @2014-8-6 @下午5:09:55
 */
@Service
public class SyncCenter {
	
	private  static final String POOL_TYPE = "SyncCenter";
	static{
		ThreadPool.createPool(POOL_TYPE, 10, 10);
	}
	@Autowired
	NodePipelineService nodePipelineservice;

	@Autowired
	@Qualifier(AppIocBean.SYNC_SENDER_RMI_IMPL)
	private ISyncSender syncSender;
	@Autowired
	SynchronizeHandler synchronizeHandler;
	public java.util.concurrent.ConcurrentLinkedQueue<SynchronizeInfo> syncTask = new ConcurrentLinkedQueue<SynchronizeInfo>();
	
	@SuppressWarnings("deprecation")
	Logger synclogger = Log.getDetachLogger(LOGTYPE.SYNC);
	
	/**
	 * 同步信息注册
	 * @param synchronizeInfo
	 */
	public void registSyncTask(SynchronizeInfo synchronizeInfo){
		syncTask.add(synchronizeInfo);
	}
	/**
	 * 同步信息注册
	 * @param synchronizeInfo
	 */
	public void registSyncTaskAndStart(SynchronizeInfo synchronizeInfo){
		syncTask.add(synchronizeInfo);
		try {
			synchronizeHandler.insertOrUpdateSynchronizeInfo(synchronizeInfo);
		} catch (ServiceException e) {
			synclogger.error("数据发送失败！",e);
		}
		startSync();
	}
	/**
	 * 同步信息注册
	 * @param synchronizeInfo
	 */
	@SuppressWarnings("unchecked")
	public void registSyncTask(String sn,NodeInfo fromNode,NodeInfo srcNode,NodeInfo toNode,String dataContent,SyncOperType syncOperType,SyncType syncType,boolean whenErrorRepeat ){
		String uuid  = UUIDGenerator.getUUID();
		SynchronizeInfo synchronizeInfo = new SynchronizeInfo();
		
		//判断是从管理节点往下同步的
		/*if (NodeInfoParam.isManaagerNode()  && srcNode.getNodeCode().equals(fromNode.getNodeCode())) {
			//对NodePipeline 通道进行处理
			if (SyncType.PILE_LINE.getCode().equalsIgnoreCase(syncType.getCode())) {
				NodePipeline pipeline=JsonUtil.toObject(dataContent,NodePipeline.class); 
				if (pipeline.getPipeType().equals(PIPE_TYPE.TONGLINKQ.getCode()) && fromNode.getId().equals(pipeline.getNodeId())) {//如果是中央的tlq，那么就根据toNode的行政区，返回一个tlq ip
				   String region=toNode.getNodeCode().substring(0,9);//获取目的节点的节点编码
				   String ip=TlqCache.regionMap.get(region);
				   try {
					NodePipeline p=nodePipelineservice.getByNodeIdAndIp(fromNode.getId(),ip);
					dataContent=JsonUtil.toJson(p);
				} catch (ServiceException e) {
					synclogger.error("数据发送失败！获取过滤处理NodePipeline信息失败",e);
				}
				}
			}else if(SyncType.PILE_LINE_LIST.getCode().equalsIgnoreCase(syncType.getCode())){
				List<NodePipeline> list=JsonUtil.toObject(dataContent,List.class); 
				List<NodePipeline> newList=new ArrayList<NodePipeline>();
				for (NodePipeline nodePipeline : list) {
					if (nodePipeline.getPipeType().equals(PIPE_TYPE.TONGLINKQ.getCode()) && fromNode.getId().equals(nodePipeline.getNodeId())) {//如果是中央的tlq，那么就根据toNode的行政区，返回一个tlq ip
						   String region=toNode.getNodeCode().substring(0,9);//获取目的节点的节点编码
						   String ip=TlqCache.regionMap.get(region);
						   try {
							NodePipeline p=nodePipelineservice.getByNodeIdAndIp(fromNode.getId(),ip);
							if (p.getId().equals(nodePipeline.getId())) {
								newList.add(p);
							}
						} catch (ServiceException e) {
							synclogger.error("数据发送失败！获取过滤处理NodePipeline信息失败",e);
						}
					}else{
						newList.add(nodePipeline);
					}
				}
				dataContent=JsonUtil.toJson(newList);
			}
		}*/
		synchronizeInfo.setDataContent(dataContent);
		synchronizeInfo.setFromNode(fromNode.getId());
		synchronizeInfo.setFromNodeName(fromNode.getNodeName());
		synchronizeInfo.setId(uuid);
		synchronizeInfo.setSrcNode(srcNode.getId());
		synchronizeInfo.setFromNodeInfo(fromNode);
		synchronizeInfo.setSrcNodeInfo(srcNode);
		synchronizeInfo.setToNodeInfo(toNode);
		synchronizeInfo.setWhenErrorRepeat(whenErrorRepeat);
		synchronizeInfo.setOperType(syncOperType.getCode());
		synchronizeInfo.setDataType(syncType.getCode());
		synchronizeInfo.setStatus(SyncStatus.WAIT.getCode());
		synchronizeInfo.setToNode(toNode.getId());
		synchronizeInfo.setToNodeName(toNode.getNodeName());
		synchronizeInfo.setSn(sn);
		synchronizeInfo.setMtime(new Date());
		synchronizeInfo.setSyncUrl(toNode.getInstallUrl());
		syncTask.add(synchronizeInfo);
		if(whenErrorRepeat){
			try {
				synchronizeHandler.insertOrUpdateSynchronizeInfo(synchronizeInfo);
			} catch (ServiceException e) {
				synclogger.error("数据发送失败！",e);
			}
		}
	}
	public void registSyncTaskAndStart(String sn,NodeInfo fromNode,NodeInfo srcNode,NodeInfo toNode,String dataContent,SyncOperType syncOperType,SyncType syncType,boolean whenErrorRepeat ){
		registSyncTask(sn, fromNode, srcNode,toNode, dataContent, syncOperType, syncType,whenErrorRepeat);
		startSync();
	}
	/**
	 * 同步信息注册
	 * @param synchronizeInfo
	 */
	public void registSyncTasks(String sn,NodeInfo fromNode,NodeInfo srcNode,List<NodeInfo> toNodes,String dataContent,SyncOperType syncOperType,SyncType syncType,boolean whenErrorRepeat ){
		for (Iterator<NodeInfo> iterator = toNodes.iterator(); iterator.hasNext();) {
			NodeInfo toNode = iterator.next();
			if(toNode.getId().equalsIgnoreCase(fromNode.getId())||toNode.getId().equalsIgnoreCase(srcNode.getId()))
				continue;
			registSyncTask(sn, fromNode,srcNode, toNode, dataContent, syncOperType, syncType,whenErrorRepeat);
		}
	}
	/**
	 * 同步信息注册
	 * @param synchronizeInfo
	 */
	public void registSyncTasksAndStart(String sn,NodeInfo fromNode,NodeInfo srcNode,List<NodeInfo> toNodes,String dataContent,SyncOperType syncOperType,SyncType syncType,boolean whenErrorRepeat ){
		for (Iterator<NodeInfo> iterator = toNodes.iterator(); iterator.hasNext();) {
			NodeInfo toNode = iterator.next();
			if(toNode.getId().equalsIgnoreCase(fromNode.getId())||toNode.getId().equalsIgnoreCase(srcNode.getId()))
				continue;
			registSyncTask(sn, fromNode,srcNode, toNode, dataContent, syncOperType, syncType,whenErrorRepeat);
		}
		startSync();
	}
	Boolean isSyncTaskRunning = false;
	/**
	 * 开始同步，按先进先出规则
	 */
	public  void startSync(){
		if(isSyncTaskRunning==false){
			synchronized (isSyncTaskRunning) {
				isSyncTaskRunning = true;
			}
		}else{
			return;
		}
		while (!syncTask.isEmpty()) {
			final SynchronizeInfo synchronizeInfo = syncTask.poll();
				if(synchronizeInfo!=null&&StringUtils.hasText(synchronizeInfo.getDataContent())){
					try {
						ThreadPool.execute(POOL_TYPE,  new Thread(new Runnable() {
							@Override
							public void run() {
								try {
									syncSender.sendData(synchronizeInfo.getSyncUrl(),synchronizeInfo);
								} catch (Exception e) {
									synclogger.error("数据发送失败！",e);
								}
							}}));
					} catch (ServiceException e) {
						synclogger.error(e);
					}
				}
		}
		isSyncTaskRunning = false;
	}
}

